package com.corn.flink.lesson2;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.concurrent.ThreadLocalRandom;

/**
 * @author : Jim Wu
 * @version 1.0
 * @function :
 * @since : 2023/2/23 15:04
 */

public class FlinkCustomSourceOperatorDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource ds = env.addSource(new MyCustomSourceOperator());
        ds.print();
        env.execute();

    }

    /**
     * 实现SourceFunction接口的run 方法 与 cancel
     */
    static class MyCustomSourceOperator implements SourceFunction<Integer> {

        private boolean flag = true;

        private ThreadLocalRandom random = ThreadLocalRandom.current();

        /**
         * 数据收集方法
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<Integer> ctx) throws Exception {
            // flag标志位表示数据的生成是否停止
            while (flag) {
                // ctx source上下文 collect可以收集生成的数据流向下游
                ctx.collect(random.nextInt(3000));
                Thread.sleep(1000);
            }
        }

        /**
         * 任务停止方法
         */
        @Override
        public void cancel() {
            flag = false;
        }
    }
}
